package com.atguigu.function;

import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.common.GmallConfig;
import com.atguigu.util.DruidPhoenixUtil;
import com.atguigu.util.PhoenixUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.util.Collection;
import java.util.Set;

/**
 * @author yhm
 * @create 2022-11-18 15:26
 */
public class DimSinkFunc extends RichSinkFunction<JSONObject> {
    DruidPooledConnection phoenixConn = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 获取连接
        phoenixConn = DruidPhoenixUtil.getPhoenixConn();
    }

    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        // 拼接sql

        String sql = createUpsertSql(value);

        PhoenixUtil.executeSql(phoenixConn, sql);

    }

    private String createUpsertSql(JSONObject value) {
        //upsert into 库名.表名  (column1,column2...)values( value1,value2 ..)
        StringBuilder stringBuilder = new StringBuilder();

        // {"is_sale":0,"sku_desc":"华为","tm_id":3,"create_time":"2022-11-18 15:22:43","price":1999,"weight":35.00,"sku_name":"华为4G+鸿蒙","sink_table":"dim_sku_info","id":36,"spu_id":12,"category3_id":86}
        String sinkTable = value.getString("sink_table");
        stringBuilder.append("upsert into ")
                .append(GmallConfig.HBASE_SCHEMA)
                .append(".")
                .append(sinkTable)
                .append("(");
        value.remove("sink_table");
        Set<String> columns = value.keySet();
        Collection<Object> values = value.values();
        stringBuilder.append(StringUtils.join(columns, ","))
                .append(") values ('")
                .append(StringUtils.join(values, "','"))
                .append("')");

        System.out.println(stringBuilder.toString());
        return stringBuilder.toString();
    }

    @Override
    public void close() throws Exception {
        super.close();
        phoenixConn.close();
    }
}
